package com.yjz.jvm.list;

import org.apache.rocketmq.common.message.Message;

import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.function.Consumer;

/**
 * @author :yuanJinZhong
 * @description: RocketMq 官网中 集合分离器
 * @date :2018/11/9-9:48
 */
public class ListSplitter implements Iterator<List<Message>> {

  private final int SIZE_LIMIT = 1024 * 1024;

  private List<Message> messages;

  private int currIndex;

  public ListSplitter(List<Message> messages) {
    this.messages = messages;
  }

  @Override
  public boolean hasNext() {
    return currIndex < messages.size();
  }

  @Override
  public List<Message> next() {
    int nextIndex = currIndex;
    int totalSize = 0;
    for (; nextIndex < messages.size(); nextIndex++) {
      Message message = messages.get(nextIndex);
      int tempSize = message.getTopic().length() + message.getBody().length;
      Map<String, String> properties = message.getProperties();
      for (Map.Entry<String, String> entry : properties.entrySet()) {
        tempSize += entry.getKey().length() + entry.getValue().length();
      }
      //for log overhead
      tempSize = tempSize + 20;
      if (tempSize > SIZE_LIMIT) {
        //it is unexpected that single message exceeds the SIZE_LIMIT
        //here just let it go, otherwise it will block the splitting process
        if (nextIndex - currIndex == 0) {
          //if the next sublist has no element, add this one and then break, otherwise just break
          nextIndex++;
        }
        break;
      }
      if (tempSize + totalSize > SIZE_LIMIT) {
        break;
      } else {
        totalSize += tempSize;
      }
    }
    List<Message> subList = messages.subList(currIndex, nextIndex);
    currIndex = nextIndex;
    return subList;
  }

}
